package ai.people.netmon.msgsearch.service.impl;

import ai.people.core.common.properties.elastic.ElasticIndex;
import ai.people.netmon.framework.domain.searchmsg.request.AccountRequestToSpeak;
import ai.people.netmon.framework.domain.searchmsg.request.DataQuery;
import ai.people.netmon.framework.domain.searchmsg.request.MessageRequest;
import ai.people.netmon.framework.domain.searchmsg.vo.AccountSpeakingLabelResponse;
import ai.people.netmon.framework.domain.searchmsg.vo.EsMessageDetailVO;
import ai.people.netmon.framework.exception.enums.AccountExceptionEnum;
import ai.people.netmon.framework.exception.enums.CommonEnum;
import ai.people.netmon.framework.model.response.Result;
import ai.people.netmon.framework.utils.AssertUtils;
import ai.people.netmon.msgsearch.mapstruct.InfoDataConverter;
import ai.people.netmon.msgsearch.service.EsSearchService;
import ai.people.netmon.msgsearch.service.SearchAfterServiceImpl;
import ai.people.netmon.utils.StringPool;
import ai.people.netmon.utils.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * es搜索服务impl
 *
 * @author yuanqinglong
 * @date 2022/1/21 11:40
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class EsSearchServiceImpl implements EsSearchService {


    @Autowired
    //private MessageSearchService messageSearchService;
    private SearchAfterServiceImpl messageSearchService;


    @Autowired
    private InfoDataConverter infoDataConverter;


    private final RestHighLevelClient elasticsearchClient;


    /**
     * 搜索
     *
     * @param messageRequest 消息请求
     * @param or
     * @return {@link Result}<{@link ?}>
     */
    @Override
    public Result search(MessageRequest messageRequest, boolean or) {
        DataQuery dataQuery = infoDataConverter.toInfoDataQuery(messageRequest);
        dataQuery.setSentiment(messageRequest.getSentiment());
        return messageSearchService.search(dataQuery, or);
    }

    /**
     * 搜索详情
     *
     * @param id       id
     * @param platform 平台
     * @return {@link EsMessageDetailVO}
     * @throws IOException ioexception
     */
    @Override
    public EsMessageDetailVO searchDetail(String id, String platform) throws IOException {
        SearchRequest request = new SearchRequest(ElasticIndex.getEsTemplateIndexName(platform));
        QueryStringQueryBuilder queryStringQueryBuilder = new QueryStringQueryBuilder("_id:" + id);
        SearchSourceBuilder source = request.source();
        log.info("查询语句:{}", source);
        source.query(queryStringQueryBuilder);
        request.source(source);
        SearchResponse search = elasticsearchClient.search(request, RequestOptions.DEFAULT);
        SearchHit[] searchHits = Optional.ofNullable(search).map(SearchResponse::getHits).map(SearchHits::getHits).orElse(new SearchHit[]{});
        AssertUtils.isTrue(searchHits.length > 0, AccountExceptionEnum.ACCOUNT_DOES_NOT_EXIST);
        SearchHit searchHit = searchHits[0];
        Map<String, Object> sources = searchHit.getSourceAsMap();
        AssertUtils.isNotNull(sources, CommonEnum.QUERY_NOT_FOUND);
        JSONObject sourceAsMap = JSON.parseObject(JSON.toJSONString(sources));
        return EsMessageDetailVO.build(sourceAsMap);

    }

    /**
     * 查询账号预警言论统计
     *
     * @param accountRequestToSpeak 帐户请求说话
     * @return {@link List}<{@link AccountSpeakingLabelResponse}>
     */
    @Override
    public List<AccountSpeakingLabelResponse> queryAccountWarningRemarksCount(AccountRequestToSpeak accountRequestToSpeak) throws IOException {
        SearchRequest searchRequest;
        // 查询ES
        if (StringUtils.isNotEmpty(accountRequestToSpeak.getPlatform())) {
            String[] split = accountRequestToSpeak.getPlatform().split(StringPool.COMMA);
            String[] platforms = new String[split.length];
            for (int i = 0; i < split.length; i++) {
                String platform = split[i];
                platforms[i] = ElasticIndex.getEsTemplateIndexName(platform);
            }
            searchRequest = new SearchRequest(platforms);
        } else {
            searchRequest = new SearchRequest(ElasticIndex.getAllIndex());
        }
        // 索引
        SearchSourceBuilder sourceBuilder = searchRequest.source();
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.must(QueryBuilders.wildcardQuery("tag.tag", "*"));
        boolQueryBuilder.must(QueryBuilders.termsQuery("user_name", accountRequestToSpeak.getAccount().split(StringPool.COMMA)));
        sourceBuilder.query(boolQueryBuilder).size(10000);
        log.info("检索语句：{}", sourceBuilder);
        SearchResponse searchResponse = elasticsearchClient.search(searchRequest, RequestOptions.DEFAULT);
        if (Objects.isNull(searchResponse) || Objects.isNull(searchResponse.getHits())) {
            return Collections.emptyList();
        }
        SearchHits hits = searchResponse.getHits();
        SearchHit[] searchHits = hits.getHits();
        Map<String, Long> accountMessageCount = Arrays.stream(searchHits).map(SearchHit::getSourceAsString).map(JSON::parseObject).collect(Collectors.groupingBy(jsonObject -> jsonObject.getString("user_name"), Collectors.counting()));
        List<AccountSpeakingLabelResponse> accountSpeakingLabelResponses = new ArrayList<>();
        accountMessageCount.forEach((account, count) -> accountSpeakingLabelResponses.add(new AccountSpeakingLabelResponse(account, count.intValue())));
        return accountSpeakingLabelResponses;
    }


}
